#include "stdafx.h"
#include "Portal.h"
#include "LocalEndPoint.h"
#include "InvokeEndpoint.h"
#include "SimpleLogger.h"
#include "DynamicBindingManager.h"

namespace RoomFW {

Portal::Portal(const char* name) :
	name_(name),
	peerObj_(),
	peerPort_(),
	peerEP_(),
	replyEndpoint_(),
	contract_(),
	messageQueue_()
{
}

Portal::~Portal(void)
{
	DynamicBindingManager::Instance().unregister(this);
}

const char* Portal::getName() const
{
	return name_.c_str();
}

SendMessageError::Reason Portal::acceptMessage(MessageID messageID,
											   IPayloadSP& payload,
											   const char* senderName,
											   IEndPointSP replyEndpoint)
{
	// Contract violations are a programming error, hence the assertion.
	ASSERT(contract_->validMessage(messageID, payload));
	SendMessageError::Reason retVal =
		(contract_->validMessage(messageID, payload) ?
			SendMessageError::SMER_NONE :
			SendMessageError::SMER_CONTRACT_VIOLATION);
	if (SendMessageError::SMER_NONE == retVal)
	{
		// Log message.
#ifndef G_NO_LV_MESSAGE
		if (G_IS_LOGGING_MESSAGE && contract_->getIsLogged())
		{
			std::stringstream logStr;
			logStr << '<' << senderName << "><" << name_ << "><" << contract_->messageName(messageID) << '>';
			if (0 == payload)
			{
				logStr << "<NULL>";
			}
			else
			{
				logStr << "<" << payload->toString() << ">";
			}
			G_LOG_0(LV_MESSAGE, logStr.str().c_str());
		}
#endif

		// Queue message.
		{
			CriticalSectionGuard guard(queueSynch_);
			messageQueue_.push(RoomMessageSP(new RoomMessage(peerPort_,
															 messageID,
															 IPayload::ReleaseOwnership(payload),
															 senderName,
															 replyEndpoint)));
		}
		this->notifyMessage();
	}
	else
	{
		if (G_IS_LOGGING_MESSAGE && contract_->getIsLogged())
		{
			std::stringstream logStr;
			logStr << '<' << senderName << "><" << name_ << "><" << contract_->messageName(messageID) << '>';
			if (0 == payload)
			{
				logStr << "<NULL>";
			}
			else
			{
				logStr << "<" << payload->toString() << ">";
			}
			G_LOG_0(LV_MESSAGE, logStr.str().c_str());
		}
	}
	return retVal;
}

void Portal::bindTo(IActiveObjectSP object, PortIndex& index)
{
	ASSERT(0 != object);
	if (0 == object)
	{
		throw std::runtime_error("object is null");
	}

	ASSERT(0 == peerObj_);
	ASSERT(0 == peerEP_);
	if ((0 != peerObj_) || (0 != peerEP_))
	{
		throw std::runtime_error("portal already bound");
	}

	peerObj_ = object;
	contract_ = peerObj_->getContractChecker(index)->getConjugate();
	if (0 == contract_)
	{
		throw std::runtime_error("null protocol");
	}
	replyEndpoint_.reset(new PortalEndPoint(this));
	if (!peerObj_->setEndPoint(index, replyEndpoint_))
	{
		throw std::runtime_error("object failed to set endpoint");
	}
	peerEP_ = peerObj_->getEndPoint(index.port());
	peerPort_ = index;
}

bool Portal::unbind(void)
{
	bool retVal = false;
	if ((0 != peerObj_) && (0 != peerEP_))
	{
		retVal = peerObj_->clearEndPoint(peerPort_);
		peerObj_.reset();
		peerEP_.reset();
		replyEndpoint_.reset();
	}
	return retVal;
}

void Portal::preSend(MessageID messageID, IPayloadSP& payload) const
{
	// Sending to an unbound portal indicates a programming error, hence the assertion.
	ASSERT(0 != peerEP_);
	if (0 == peerEP_)
	{
		throw SendMessageError(
				name_.c_str(),
				peerPort_,
				messageID,
				payload,
				SendMessageError::SMER_PORT_NOT_BOUND);
	}
}

void Portal::finalSend(MessageID messageID, IPayloadSP& payload, IEndPointSP replyEndpoint) const
{
	SendMessageError::Reason reason =
		peerEP_->acceptMessage(messageID, payload, (G_IS_LOGGING_MESSAGE ? name_.c_str() : ""), replyEndpoint);
	if (SendMessageError::SMER_NONE != reason)
	{
		throw SendMessageError(
				name_.c_str(),
				peerPort_,
				messageID,
				payload,
				reason);
	}
}

void Portal::sendMessage(MessageID messageID, IPayloadSP payload)
{
	// Catch SendMessageError's to fill in port and message text and re-throw
	try
	{
		// Check for errors.
		this->preSend(messageID, payload);

		// Send the message.
		ASSERT(0 != replyEndpoint_); // by this class.
		this->finalSend(messageID, payload, replyEndpoint_);
	}
	catch (SendMessageError& e)
	{
		e.setObjectName(name_.c_str());
		if (0 != peerObj_)
		{
			e.setPortIndexText(peerObj_->getPortName(peerPort_.port()));
		}
		if (0 != contract_)
		{
			e.setMessageIdText(contract_->messageName(messageID));
		}
		throw e;
	}
}

RoomMessageSP Portal::invoke(uint32_t timeout, MessageID messageID, IPayloadSP payload, bool throwOnTimeout)
{
	// Catch SendMessageError's to fill in port and message text and re-throw
	RoomMessageSP retVal;
	try
	{
		// Check for errors.
		this->preSend(messageID, payload);

		// Send the message.
		std::stringstream logStr;
		logStr << '<' << name_ << '>';
		InvokeEndpointSP invokeEndpoint(new InvokeEndpoint(peerPort_, contract_, logStr.str().c_str()));
		this->finalSend(messageID, payload, invokeEndpoint);

		// Wait for reply and handle error.
		retVal = invokeEndpoint->wait(timeout);
		if ((0 == retVal) && throwOnTimeout)
		{
			throw SendMessageError(
					this->name_.c_str(),
					peerPort_,
					messageID,
					payload,
					SendMessageError::SMER_INVOKE_TIMEOUT);
		}
	}
	catch (SendMessageError& e)
	{
		e.setObjectName(name_.c_str());
		if (0 != peerObj_)
		{
			e.setPortIndexText(peerObj_->getPortName(peerPort_.port()));
		}
		if (0 != contract_)
		{
			e.setMessageIdText(contract_->messageName(messageID));
		}
		throw e;
	}
	return retVal;
}

bool Portal::hasMessage(void)
{
	bool retVal = false;
	{
		CriticalSectionGuard guard(queueSynch_);
		retVal = !messageQueue_.empty();
	}
	return retVal;
}

RoomMessageSP Portal::getMessage(void)
{
	RoomMessageSP retVal;
	{
		CriticalSectionGuard guard(queueSynch_);
		if (!messageQueue_.empty())
		{
			retVal = messageQueue_.front();
			messageQueue_.pop();
		}
	}
	return retVal;
}

DynamicBindingSP Portal::dynamicBind(const char* dynamicPortName, ContractTypeId contractTypeId, bool isConjugate)
{
	return DynamicBindingManager::Instance().dynamicBind(this, dynamicPortName, contractTypeId, isConjugate);
}

bool Portal::getIsLogged() const
{
	bool retVal = false;
	if (contract_)
	{
		retVal = contract_->getIsLogged();
	}
	return retVal;
}

void Portal::setIsLogged(bool isLogged)
{
	if (contract_)
	{
		contract_->setIsLogged(isLogged);
	}
}

IContractCheckerSPC Portal::getContractChecker()
{
	return contract_;
}


Win32Portal::Win32Portal(const char* name, HWND hWnd, UINT messageId, WPARAM wParam, LPARAM lParam) :
	Portal(name),
	hWnd_(hWnd), 
	messageId_(messageId),
	wParam_(wParam),
	lParam_(lParam)
{
}

void Win32Portal::setNotifyInfo(HWND hWnd, UINT messageId, WPARAM wParam, LPARAM lParam)
{
	hWnd_ = hWnd;
	messageId_ = messageId;
	wParam_ = wParam;
	lParam_ = lParam;
}

void Win32Portal::notifyMessage(void)
{
	::PostMessage(hWnd_, messageId_, wParam_, lParam_);
}

EventPortal::EventPortal(const char* name) :
	Portal(name),
	messageEvent_()
{
}

bool EventPortal::initialize(void)
{
	return messageEvent_.create(true);
}

bool EventPortal::wait(uint32_t timeout)
{
	VERIFY(messageEvent_.reset());
	return (this->hasMessage() ||
			(messageEvent_.wait(timeout) && this->hasMessage()));
}

void EventPortal::notifyMessage(void)
{
	VERIFY(messageEvent_.set());
}

CallbackPortal::CallbackPortal(const char* name, ICallback* callback) :
	Portal(name),
	callback_(callback),
	callbackSynch_()
{
}

void CallbackPortal::setCallback(ICallback* callback)
{
	CriticalSectionGuard g(callbackSynch_);
	callback_ = callback;
}

void CallbackPortal::notifyMessage(void)
{
	ICallback* callback = 0;
	{
		CriticalSectionGuard g(callbackSynch_);
		callback = callback_;
	}
	ASSERT(0 != callback);
	if (0 != callback)
	{
		callback_->notifyMessage(this);
	}
}

PortalQueue::PortalQueue() :
	readyPortals_(),
	queueSynch_(),
	messageEvent_()
{
	VERIFY(messageEvent_.create(true));
}

bool PortalQueue::wait(uint32_t timeout)
{
	VERIFY(messageEvent_.reset());
	return (this->hasMessage() ||
			(messageEvent_.wait(timeout) && this->hasMessage()));
}

CallbackPortal* PortalQueue::pop()
{
	CallbackPortal* retVal = 0;
	CriticalSectionGuard guard(queueSynch_);
	if (!readyPortals_.empty())
	{
		retVal = readyPortals_.front();
		readyPortals_.pop();
	}
	return retVal;
}

bool PortalQueue::hasMessage()
{
	bool retVal = false;
	{
		CriticalSectionGuard guard(queueSynch_);
		retVal = !readyPortals_.empty();
	}
	return retVal;
}

void PortalQueue::notifyMessage(CallbackPortal* portal)
{
	CriticalSectionGuard guard(queueSynch_);
	readyPortals_.push(portal);
	VERIFY(messageEvent_.set());
}


}
